#include "CEntranceMqttClient.h"
#include "CControlManage.h"
NETWORKSERVICE_BEGIN_NAMESPACE

// 数据回调
static void publishData(string topics,string &msg,void *m_pData)
{
	CEntranceMqttClient *p = (CEntranceMqttClient*)m_pData;
    qDebug("publishData:%s:%s",topics.c_str(),msg.c_str());
	if(p){
		p->__publishData(topics, msg);
	}
}

// 连接状态回调
static void conStatus(bool status,void *m_pData)
{
	CEntranceMqttClient *p = (CEntranceMqttClient*)m_pData;
    qDebug("conStatus:%d",status);
	if(p){
		p->__conStatus(status);
	}
}

CEntranceMqttClient::CEntranceMqttClient()
{
	m_conStatus = false;
}

CEntranceMqttClient::~CEntranceMqttClient()
{
	stop();
}

// 启动
bool CEntranceMqttClient::start()
{
	return m_mqttClient.start();
}
// 停止
void CEntranceMqttClient::stop()
{
	m_mqttClient.stop();
}

// 初始化产生
void CEntranceMqttClient::init(string url,struct mg_mqtt_opts *sOpts,struct mg_tls_opts *sTlsOpts,int reconnectionTime)
{
	m_mqttClient.init(url, sOpts, sTlsOpts, reconnectionTime);
	m_back.conStatus = conStatus;
	m_back.publishData = publishData;
	m_mqttClient.setBack(&m_back,this);
}

// 订阅
bool CEntranceMqttClient::subscribes(string topics,int qos)
{
	uint16_t ret = 0;
	if(topics.empty())
		return false;
	m_subscribesMap[topics] = qos;
	if(!m_conStatus){
		return true;
	}
	ret = m_mqttClient.subscribes(topics.c_str(), qos);
    qDebug("subscribes:%s",topics.c_str());
	return ret == 0 ? false : true;

}

// 取消订阅
uint16_t CEntranceMqttClient::unSubscribes(string topics)
{
	const char *p = topics.c_str();
	if(m_subscribesMap.find(topics) != m_subscribesMap.end()){
		m_subscribesMap.erase(topics);
	}
	return m_mqttClient.unSubscribes((char **)&p, 1);
}

// 发布消息
uint16_t CEntranceMqttClient::publish(string topics,string &data,int qos,bool retain)
{
	return m_mqttClient.publish(topics.c_str(), data, qos, retain);
}

// 连接状态回调
void CEntranceMqttClient::__conStatus(bool status)
{
	if(status){
		MAP<string,int>::iterator itr;
		for(itr = m_subscribesMap.begin(); itr!= m_subscribesMap.end(); ++itr){
			m_mqttClient.subscribes(itr->first.c_str(), itr->second);
            qDebug("subscribes:%s",itr->first.c_str());
		}
	}
	m_conStatus = status;
}




typedef struct _tThreadData{
	XJSON_FUNCI
	string topics;
	string msg;
	XPACK( O(topics, msg));
}T_ThreadData;

static list<T_ThreadData> listThreadData;
static QMutex     ThreadDataMutex;					// 互斥锁

static void __dataHandle(void)
{
	static bool isRun = false;
	if(isRun){
        qDebug("mqtt __dataHandle() is busy");
        return;
	}
	isRun = true;

	while(true){
        if(listThreadData.empty()){
            QThread::msleep(10000);
			continue;
		}
		T_ThreadData data;
		ThreadDataMutex.lock();
		data = *listThreadData.begin();
		listThreadData.erase(listThreadData.begin());
        ThreadDataMutex.unlock();

		// 消息路由
		struct tRoute route;
		// 消息来源
		route.type = ROUTE_MQTT_SERVICE;
		// mqtt主题
		route.mqttTopics = data.topics;

		struct _tCmdBaseData head;
		head.loadJsonString(data.msg);
		const T_NetConfig& netConfig = CNetConfig::getConfig();
        string md5str = controlManage::strMd5(controlManage::strMd5(head.sendMsgId + std::to_string(head.timestamp)));
        qDebug("md5str:%s", md5str.c_str());
        qDebug("head.strSignature:%s", head.signature.c_str());

		route.isAuthorizationSuccess = head.signature == md5str;

        bool ret = controlManage::runMqttHandle(&head, data.msg, &route);

	}


	isRun = false;
    return;
}

// 数据回调
void CEntranceMqttClient::__publishData(string topics,string &msg)
{
	ThreadDataMutex.lock();
	T_ThreadData data = {topics, msg};
	listThreadData.push_back(data);
    ThreadDataMutex.unlock();

	static bool isStart = false;
	if(isStart){
		return;
	}
	isStart = true;

    std::thread t(__dataHandle);
    t.detach();


}
NETWORKSERVICE_END_NAMESPACE
